package org.apache.flink.kubernetes.entrypoint;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils;
import org.apache.flink.runtime.entrypoint.DynamicParametersConfigurationParserFactory;
import org.apache.flink.runtime.entrypoint.master.DefaultMasterComponentFactory;
import org.apache.flink.runtime.entrypoint.master.MasterComponentFactory;
import org.apache.flink.runtime.resourceconsumer.runner.ResourceConsumerServiceLeadershipRunnerFactory;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;

/**
 * @类名: KubernetesClusterEntrypoint
 * @说明: 
 *
 * @author   leehom
 * @Date	 2024年8月15日 下午3:04:26
 * 修改记录：
 *
 * @see 	 
 */
public class KubernetesClusterEntrypoint extends ClusterEntrypoint {

    public KubernetesClusterEntrypoint(Configuration configuration) {
        super(configuration);
    }

    @Override
    protected MasterComponentFactory
            createMasterComponentFactory(Configuration configuration) {
        return DefaultMasterComponentFactory.createMasterComponentFactory(
        		ResourceConsumerServiceLeadershipRunnerFactory.INSTANCE,
                KubernetesResourceManagerFactory.getInstance());
    }

    public static void main(String[] args) {
        // startup checks and logging
        EnvironmentInformation.logEnvironmentInfo(
                LOG, KubernetesClusterEntrypoint.class.getSimpleName(), args);
        SignalHandler.register(LOG);
        JvmShutdownSafeguard.installAsShutdownHook(LOG);

        final Configuration dynamicParameters =
                ClusterEntrypointUtils.parseParametersOrExit(
                        args,
                        new DynamicParametersConfigurationParserFactory(),
                        KubernetesClusterEntrypoint.class);
        // 构建集群入口点
        final ClusterEntrypoint entrypoint =
                new KubernetesClusterEntrypoint(
                        KubernetesEntrypointUtils.loadConfiguration(dynamicParameters));
        // run
        ClusterEntrypoint.runClusterEntrypoint(entrypoint);
    }
}
